package com.atguigu.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class Hbase_DML {
    //获取hbase连接
    public static Connection connection = Hbase_Connect_Single.connection;

    //增加数据
    public static void putCell(String nameSpace, String tableName, String rowKey, String family, String cloumn, String value) throws IOException {

        //1.通过Hbase连接获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //2.调用Put往Hbase中写入数据

        //创建Put对象
        Put put = new Put(Bytes.toBytes(rowKey));

        //在put对象中添加要写的数据
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(cloumn), Bytes.toBytes(value));

        table.put(put);

        //关闭Table
        table.close();
    }


    //get获取某一行或某一个cell数据
    public static void getCells(String nameSpace, String tableName, String rowKey, String family, String cloumn) throws IOException {
        //1.通过Hbase连接获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //2.调用get方法获取数据

        //创建Get对象
        Get get = new Get(Bytes.toBytes(rowKey));

        //获取所有的版本
        get.readAllVersions();

        //获取某个列族的数据
        get.addFamily(Bytes.toBytes(family));
//        获取某一列的数据
//        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(cloumn));

        //获取结果集
        Result result = table.get(get);

        //遍历Result获取到每一个Cell
        for (Cell cell : result.rawCells()) {
            System.out.println(Bytes.toString(CellUtil.cloneRow(cell)) + "-" + Bytes.toString(CellUtil.cloneFamily(cell)) + "-" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "-" + Bytes.toString(CellUtil.cloneValue(cell)));
        }

        //打印
//        System.out.println(Bytes.toString(result.value()));

        table.close();
    }
    //删除数据
    public static void deleteCells(String nameSpace, String tableName, String rowKey, String family, String cloumn) throws IOException {
        //1.获取table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //2.调取delete方法删除数据

        //创建Delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        //删除单个版本(最新版本)
//        delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(cloumn));

        //删除所有版本
        delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(cloumn));

        table.delete(delete);

        table.close();
    }

    //使用scan扫描数据
    public static void scanTable(String nameSpace, String tableName,String startRow,String stopRow) throws IOException {
        //1.获取Table对象
        Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));

        //2.调取scan获取表的数据

        //创建Scan对象
        Scan scan = new Scan();

        //读取超过维护版本的数据
//        scan.setRaw(true);

        //读取所有版本的数据
//        scan.readAllVersions();

        //读取指定版本个数的数据
//        scan.readVersions(6);

        //根据rowkey范围获取数据  第二个boolean类型的参数代表包含不包含本行，true包含 false反之
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow),true);

        ResultScanner scanner = table.getScanner(scan);

        //遍历ResultScanner获取所有的Result结果集
        for (Result result : scanner) {
            //遍历Result获取到每一个cell
            for (Cell cell : result.rawCells()) {
                System.out.println(Bytes.toString(CellUtil.cloneRow(cell)) + "-" + Bytes.toString(CellUtil.cloneFamily(cell)) + "-" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "-" + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        table.close();
    }

    public static void main(String[] args) throws IOException {
        //写入数据
//        putCell("bigdata", "student", "1001", "info", "name", "panjinlian");
//        putCell("bigdata", "student", "1001", "info", "age", "18");
//        putCell("bigdata", "student", "1001", "info", "age", "19");
//        putCell("bigdata", "student", "1001", "info", "age", "20");
//        putCell("bigdata", "student", "1001", "info", "age", "21");
//        putCell("bigdata", "student", "1001", "info", "age", "22");
//        putCell("bigdata", "student", "1001", "info", "age", "23");
//        putCell("bigdata", "student", "1002", "info", "age", "18");
//        putCell("bigdata", "student", "1002", "info", "age", "19");
//        putCell("bigdata", "student", "1002", "info", "age", "20");
//        putCell("bigdata", "student", "1002", "info", "name", "wudalang");
//        putCell("bigdata", "student", "1003", "info", "age", "19");
//        putCell("bigdata", "student", "1003", "info", "name", "ximenqing");
//        putCell("bigdata", "student", "1004", "info", "age", "33");
//        putCell("bigdata", "student", "1004", "info", "name", "wusong");
//        putCell("bigdata", "student", "1005", "info", "age", "29");
//        putCell("bigdata", "student", "1005", "info", "name", "sunerniang");

        //获取数据
//        getCells("bigdata", "student", "1001", "info", "age");

        //删除数据
//        deleteCells("bigdata", "student", "1001", "info", "age");

//        getCells("bigdata", "student", "1001", "info", "age");

        //扫描数据
        scanTable("bigdata", "student", "1001", "1003");

        //关闭Hbase连接
        Hbase_Connect_Single.closeConnect();
    }
}
